package top.birdhk.TestAPI.source;

import com.sun.org.apache.bcel.internal.generic.NEW;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import top.birdhk.TestAPI.beans.SensorReading;

import java.sql.SQLTransactionRollbackException;
import java.util.HashMap;
import java.util.Random;

/**
 * 自定义数据源
 */
public class SourceUDF {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<SensorReading> dataStream = env.addSource(new MySensorSource());

        dataStream.print();

        env.execute();


    }


    public static class MySensorSource implements SourceFunction<SensorReading> {

        // 定义标志位，控制数据产生与停止
        private boolean running = true;

        @Override
        public void run(SourceContext<SensorReading> sourceContext) throws Exception {

            // 随机数生成器
            Random random = new Random();

            // 设置10个传感器的初始温度
            HashMap<String, Double> sensorMap = new HashMap<String, Double>();

            for (int i = 0; i < 10; i++) {
                sensorMap.put("sensor_" + (i+1),60 + random.nextGaussian() * 20);
            }

            while (running){
                for (String sensorId : sensorMap.keySet()){
                    // 在当前温度随机波动
                    Double aDouble = sensorMap.get(sensorId) + random.nextGaussian();
                    sensorMap.put(sensorId,aDouble);
                    sourceContext.collect(new SensorReading(sensorId,System.currentTimeMillis(),aDouble));
                }
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

}
